package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.function.KeywordUDTF;
import com.atguigu.gmall.realtime.beans.KeywordBean;
import com.atguigu.gmall.realtime.utils.MyClickhouseUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author Felix
 * @date 2023/4/19
 * 搜索关键词聚合统计
 * 需要启动的进程
 *      zk、kafka、flume、DwdTrafficBaseLogSplit、DwsTrafficSourceKeywordPageViewWindow
 */
public class DwsTrafficSourceKeywordPageViewWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //1.3 指定表执行的环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //1.4 注册udtf函数到表执行环境中
        tableEnv.createTemporarySystemFunction("ik_analyze",KeywordUDTF.class);

        //TODO 2.检查点相关的设置(略)
        //TODO 3.从kafka的页面日志主题中读取数据 创建动态表 并执行Watermark以及提取事件时间字段
        String topic = "dwd_traffic_page_log";
        String groupId = "dws_traffic_keyword_group";
        tableEnv.executeSql("CREATE TABLE page_log (\n" +
            "    common map<string,string>,\n" +
            "    page map<string,string>,\n" +
            "    ts bigint,\n" +
            "    rowtime as TO_TIMESTAMP(FROM_UNIXTIME(ts/1000)),\n" +
            "    WATERMARK FOR rowtime AS rowtime - INTERVAL '3' SECOND  \n" +
            ")" + MyKafkaUtil.getKafkaDDL(topic,groupId));

        // tableEnv.executeSql("select * from page_log").print();
        //TODO 4.过滤出搜索行为
        Table searchTable = tableEnv.sqlQuery("select\n" +
            "    page['item'] fullword,\n" +
            "    rowtime \n" +
            "from page_log where page['last_page_id']='search' \n" +
            "and page['item_type']='keyword' and page['item'] is not null");
        tableEnv.createTemporaryView("search_table",searchTable);
        // tableEnv.executeSql("select * from search_table").print();

        //TODO 5.使用自定义的UDTF函数对搜索内容进行分词  并将分词结果和表中的其它字段进行关联
        Table keywordTable = tableEnv.sqlQuery("SELECT keyword,rowtime\n" +
            "FROM search_table,LATERAL TABLE(ik_analyze(fullword)) t(keyword)");
        tableEnv.createTemporaryView("keyword_table",keywordTable);
        // tableEnv.executeSql("select * from keyword_table").print();

        //TODO 6.分组、开窗、聚合计算
        Table reduceTable = tableEnv.sqlQuery("select \n" +
            "\tDATE_FORMAT(TUMBLE_START(rowtime, INTERVAL '10' second), 'yyyy-MM-dd HH:mm:ss') stt,\n" +
            "\tDATE_FORMAT(TUMBLE_END(rowtime, INTERVAL '10' second), 'yyyy-MM-dd HH:mm:ss') edt,\n" +
            "    keyword,\n" +
            "    count(*) keyword_count,\n" +
            "    UNIX_TIMESTAMP()*1000 ts\n" +
            "from keyword_table group by keyword,TUMBLE(rowtime, INTERVAL '10' second)");
        tableEnv.createTemporaryView("reduce_table",reduceTable);
        // tableEnv.executeSql("select * from reduce_table").print();

        //TODO 7.将动态表转换为流
        DataStream<KeywordBean> keywordDS = tableEnv.toDataStream(reduceTable, KeywordBean.class);
        keywordDS.print(">>>>");

        //TODO 8.将流中的数据写到Clickhouse表中
        /*keywordDS.addSink(new SinkFunction<KeywordBean>() {
            @Override
            public void invoke(KeywordBean keywordBean, Context context) throws Exception {
                //注册驱动
                Class.forName(GmallConfig.CLICKHOUSE_DRIVER);
                //建立连接
                Connection conn = DriverManager.getConnection(GmallConfig.CLICKHOUSE_URL);
                //获取数据库操作对象
                String sql = "insert into dws_traffic_keyword(stt,edt,keyword,keyword_count,ts) values(?,?,?,?,?)";
                PreparedStatement ps = conn.prepareStatement(sql);
                ps.setObject(1,keywordBean.getStt());
                ps.setObject(2,keywordBean.getEdt());
                ps.setObject(3,keywordBean.getKeyword());
                ps.setObject(4,keywordBean.getKeyword_count());
                ps.setObject(5,keywordBean.getTs());
                //执行SQL
                ps.execute();
                //释放资源
                ps.close();
                conn.close();
            }
        });*/
        /*keywordDS.addSink(JdbcSink.<KeywordBean>sink(
            "insert into dws_traffic_keyword(stt,edt,keyword,keyword_count,ts) values(?,?,?,?,?)",
            new JdbcStatementBuilder<KeywordBean>() {
                @Override
                public void accept(PreparedStatement ps, KeywordBean keywordBean) throws SQLException {
                    //将流中对象的属性 赋值给问号占位符
                    ps.setObject(1,keywordBean.getStt());
                    ps.setObject(2,keywordBean.getEdt());
                    ps.setObject(3,keywordBean.getKeyword());
                    ps.setObject(4,keywordBean.getKeyword_count());
                    ps.setObject(5,keywordBean.getTs());
                }
            },
            new JdbcExecutionOptions.Builder()
                .withBatchSize(5)
                .build(),
            new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withDriverName(GmallConfig.CLICKHOUSE_DRIVER)
                .withUrl(GmallConfig.CLICKHOUSE_URL)
                .build()
        ));*/

        keywordDS.addSink(
            MyClickhouseUtil.getSinkFunction("insert into dws_traffic_keyword_page_view_window values(?,?,?,?,?)")
        );

        env.execute();
    }
}
